package com.xr.redis.pubsub.service.impl;

import com.xr.redis.pubsub.enums.RedisPubSubImplementType;
import com.xr.redis.pubsub.model.Message;
import com.xr.redis.pubsub.service.IPubSubService;
import com.xr.redis.pubsub.service.IPubSubServiceAdapter;
import com.xr.redis.pubsub.service.listener.AbstractPubSubListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Arnold Yand
 * @summary redis发布订阅
 * @time 2018/8/4 13:51
 */
public class RedisPubsubService implements IPubSubService{

  protected Logger logger = LoggerFactory.getLogger(RedisPubsubService.class);

  private int bufferSize = 100000; // default value

  private IPubSubServiceAdapter adapter;

  /**
   * 记录每个频道与监听器的对应关系
   * 一个频道可能对应多个监听器
   */
  private ConcurrentHashMap<String, List<AbstractPubSubListener>> channelMapToListener = new ConcurrentHashMap<>();

  /**
   * 根据implement方式，初始化不同的处理逻辑
   */
  public void init() throws Exception {
    switch (this.implement) {
      case LIST:{
        adapter = new ListImplementAdapter(this.getJedis(), channelMapToListener);
        adapter.init();
        break;
      }

      default:{
        adapter = new PubSubImplementAdapter(this.getJedis(), channelMapToListener);
        adapter.init();
        break;
      }
    }
  }

  @Override
  public void publish(String channel, Message message) throws Exception {
    if(     channel == null ||
            channel.trim().length() == 0 ||
            message == null
            ){
      throw new IllegalStateException("发布消息失败，参数不正确。");
    }

    this.adapter.publish(channel, message);
    logger.info("发布消息成功，message id : {}", message.getMsgId());
  }

  @Override
  public void subscribe(String channel, AbstractPubSubListener listener) throws Exception {
    if(     channel == null ||
            channel.trim().length() == 0 ||
            listener == null
            ){
      throw new IllegalStateException("订阅频道失败，参数不正确。");
    }

    // 记录订阅关系
    List<AbstractPubSubListener> listeners = channelMapToListener.get(channel);
    if(listeners == null){
      listeners = new ArrayList<AbstractPubSubListener>();
    }
    listeners.add(listener);

    adapter.subscribe(channel);

    logger.info("订阅频道：{}，成功", channel);
  }

  @Override
  public void unSubscribe(String channel, AbstractPubSubListener listener) throws Exception {
    List<AbstractPubSubListener> listeners = channelMapToListener.getOrDefault(channel, Collections.EMPTY_LIST);
    AbstractPubSubListener lstn = null;
    Iterator<AbstractPubSubListener> it = listeners.iterator();
    while (it.hasNext()) {
      lstn = it.next();
      if(lstn.getListenerId().equals(lstn.getListenerId())){
        it.remove();
        break;
      }
    }
  }

  @Autowired
  private ShardedJedisPool shardedJedisPool;

  /**
   * 支持方式，默认PUB_SUB
   */
  private RedisPubSubImplementType implement = RedisPubSubImplementType.PUB_SUB;

  private Jedis jedisClient;

  private Jedis getJedis(){
    if(jedisClient == null){
      ShardedJedis shardedJedis = shardedJedisPool.getResource();
      if(shardedJedis == null){
        throw new IllegalStateException("无法获取到redis连接");
      }

      Jedis[] jedis = new Jedis[]{};
      jedis = shardedJedis.getAllShards().toArray(jedis);
      if(jedis == null || jedis.length < 1){
        throw new IllegalStateException("无法加载Jedis");
      }

      jedisClient = jedis[0];
    }

    return jedisClient;
  }

  public int getBufferSize() {
    return bufferSize;
  }

  public void setBufferSize(int bufferSize) {
    this.bufferSize = bufferSize;
  }

  public RedisPubSubImplementType getImplement() {
    return implement;
  }

  public void setImplement(RedisPubSubImplementType implement) {
    this.implement = implement;
  }
}